package simple.demo.zk;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZKUtil;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;

public class ZkBarrier {
	
	private final String name;
	private final int barrierSize;
	private ZooKeeper zk;
	private final Object mutex = new Object();
	private CountDownLatch connectionLatch = new CountDownLatch(1);
	private String nodePath;
	
	private final Watcher watcher = new Watcher() {
		@Override
		public void process(WatchedEvent event) {
			if(event.getState() == Watcher.Event.KeeperState.SyncConnected)
				connectionLatch.countDown();
			
			synchronized(mutex) {
				mutex.notify();
			}
		}
	};
	
	public ZkBarrier(String name, int barrierSize) {
		this.name = name;
		this.barrierSize = barrierSize;
	}
	
	void connectTo(String host) throws IOException {
		zk = new ZooKeeper(host,3000, watcher);
	}
	
	void createRoot() throws InterruptedException, KeeperException {
		zk.create(ZkBarrierHelper.getBarrierRoot(), null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}
	
	void cleanUp() throws InterruptedException, KeeperException {
		ZKUtil.deleteRecursive(zk, ZkBarrierHelper.getBarrierRoot());
	}
	
	void stop() throws InterruptedException {
		if(zk != null) {
			zk.close();
		}
		connectionLatch = new CountDownLatch(1); //reset
	}
	
	void waitForConnection() throws InterruptedException {
		connectionLatch.await();
	}
	
	boolean enter() throws KeeperException, InterruptedException {
		nodePath = zk.create(ZkBarrierHelper.getBarrierPath(name), null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
		System.out.println("Node path = " + nodePath);
		
		while(true) {
			synchronized (mutex) {
				List<String> list = zk.getChildren(ZkBarrierHelper.getBarrierRoot(), true);
				
				//System.out.println(name + " - list size = " + list.size()); 
				if(list.size() < barrierSize)
					mutex.wait();
				else
					return true;
			}
		}
	}
	
	boolean leave() throws InterruptedException, KeeperException {
		zk.delete(nodePath, 0);
		
		while(true) {
			synchronized (mutex) {
				List<String> list = zk.getChildren(ZkBarrierHelper.getBarrierRoot(), true);
				if(list.size() > 0)
					mutex.wait();
				else
					return true;
			}
		}
	}
	
	public String getName() {
		return name;
	}
}
